#include "act_watch.hpp"
#include "../action_constants.hpp"


using etcdserverpb::RangeRequest;
using etcdserverpb::RangeResponse;
using etcdserverpb::WatchCreateRequest;

namespace etcdv3 {

// init 1, write 2, read 3, done 4
AsyncWatchAction::AsyncWatchAction(const ActionParameters& param)
    : Action(param) {
    isCancelled = false;

    WatchRequest watch_req;
    WatchCreateRequest watch_create_req;
    watch_create_req.set_key(parameters.key);
    watch_create_req.set_prev_kv(true);
    watch_create_req.set_start_revision(parameters.revision);

    if (parameters.withPrefix) {
        std::string range_end(parameters.key);
        int ascii = (int)range_end[range_end.length() - 1];
        range_end.back() = ascii + 1;
        watch_create_req.set_range_end(range_end);
    }

    watch_req.mutable_create_request()->CopyFrom(watch_create_req);

    context.set_initial_metadata_corked(true);
    stream = parameters.watch_stub->AsyncWatch(&context, &cq_, (void*)1);

    stream->Write(watch_req, (void*)2);
    stream->Read(&reply, (void*)3);
}

void AsyncWatchAction::waitForResponse() {
    void* got_tag;
    bool ok = false;

    while (cq_.Next(&got_tag, &ok)) {
        if (ok == false) {
            break;
        }
        if (got_tag == (void*)4) {
            isCancelled = true;
        } else if (got_tag == (void*)3) { // read tag
            if (reply.events_size()) {
                stream->WritesDone((void*)4);
            } else {
                stream->Read(&reply, (void*)3);
            }
        }
    }
}

void AsyncWatchAction::CancelWatch() {
    if (isCancelled == false) {
        stream->WritesDone((void*)4);
    }
}

void AsyncWatchAction::waitForResponse(std::function<void(const etcd::Response&)> callback) {
    void* got_tag;
    bool ok = false;

    while (cq_.Next(&got_tag, &ok)) {
        if (ok == false) {
            break;
        }
        if (got_tag == (void*)4) {
            isCancelled = true;
        } else if (got_tag == (void*)3) { // read tag
            if (reply.events_size()) {
                callback(ParseResponse());
            }
            stream->Read(&reply, (void*)3);
        }
    }
}

AsyncWatchResponse AsyncWatchAction::ParseResponse() {
    AsyncWatchResponse watch_resp;
    if (!status.ok()) {
        watch_resp.set_error_code(status.error_code());
        watch_resp.set_error_message(status.error_message());
    } else {
        watch_resp.ParseResponse(reply);
    }
    return watch_resp;
}
    
}